/**
 * Copyright (C) 2010, 2011 Neofonie GmbH
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package eu.dicodeproject.analysis.histogram;

import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;

import java.io.IOException;
import java.util.Calendar;

/**
 * Reads creation dates from HBase and emits years, years+months, years+months+day
 * and individual houts separately as keys with value 1.
 */
public class DateHistogramMapper extends TableMapper<Text, IntWritable> {

  /** Set of patterns with associated key for later output. */
  public static enum ErrorCases {
    DATENULL
  }

  private String splitExpression = DateHistogramDriver.splitExpression;

  /** Text and IntWritable for reuse */
  private Text resultKey = new Text();
  private IntWritable one = new IntWritable(1);

  /** Calendar to translate from long to dates */
  private Calendar cal = Calendar.getInstance();

  /**
   * Map
   */
  @Override
  protected void map(final ImmutableBytesWritable row, final Result values, final Context context)
      throws IOException, InterruptedException {

    // extract the data
    final byte[] creationDateBytes = values.value(); // Driver must specify a setup where only one value is in the table!

    // non-existent creationDate values are ignored
    if (creationDateBytes == null) {
      context.getCounter(ErrorCases.DATENULL).increment(1);
      return;
    }

    cal.setTimeInMillis(Bytes.toLong(creationDateBytes));

    // year only
    String year = String.valueOf(cal.get(Calendar.YEAR));
    resultKey.set("year"+splitExpression+year);
    context.write(resultKey, one);

    // year and month
    String month = String.valueOf(cal.get(Calendar.MONTH) + 1); // Calendar.get returns 0 for January
    if (month.length() == 1) {
      month = "0"+month;
    }
    resultKey.set("yearMonth"+splitExpression+year+"-"+month);
    context.write(resultKey, one);

    // year, month and day
    String day = String.valueOf(cal.get(Calendar.DAY_OF_MONTH));
    if (day.length() == 1) {
      day = "0"+day;
    }
    resultKey.set("yearMonthDay"+splitExpression+year+"-"+month+"-"+day);
    context.write(resultKey, one);

    // hour only (what time of the day are people tweeting?)
    String hour = String.valueOf(cal.get(Calendar.HOUR_OF_DAY));
    if (hour.length() == 1) {
      hour = "0"+hour;
    }
    resultKey.set("hour"+splitExpression+hour);
    context.write(resultKey, one);
  }
}
